﻿/*
   Copyright 2012 Dmitry Bratus

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
*/

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Reflection;
using Flower.Actors;
using Flower.Logging;
using Flower.Services;
using Flower.Services.Data;
using Flower.Workflow;
using Spring.Core.IO;
using Flower.Directory.Util;
using System.Threading;
using Spring.Context;
using Spring.Context.Support;
using System.Transactions;
using Assembly = System.Reflection.Assembly;
using Process = Flower.Services.Data.Process;
using SortOrder = Flower.Services.Data.SortOrder;
using System.ServiceModel;
using Spring.Objects.Factory;
using Spring.Aop.Framework;

namespace Flower.Client
{
    /// <summary>
    /// The directory client implementation.
    /// </summary>
    public sealed class FlowerClient : IFlowerClient
    {
        #region InitializationContext

        private class InitializationContext : IInitializationContext, IProcessesManager
        {
            private readonly FlowerClient _client;
            private readonly WorkflowLogAdapter _log;

            public InitializationContext(FlowerClient client, Log log)
            {
                _client = client;
                _log = new WorkflowLogAdapter(log);
            }

            public IServicesProvider ServicesProvider
            {
                get { return _client; }
            }

            public ILockManager LockManager
            {
                get { return _client; }
            }

            public IProcessesManager ProcessesManager
            {
                get { return this; }
            }

            ILog IInitializationContext.Log
            {
                get { return _log; }
            }

            public ProcessStartResult<TOut> StartProcess<TIn, TOut>(string workflow, TIn args)
            {
                return _client.StartProcess<TIn, TOut>(workflow, args, _log.Log);
            }
        }

        #endregion

        private static readonly Log Log = LogManager.CreateLog(typeof(FlowerClient).Name);

        private static readonly object[] EmptyObjectArray = new object[0];
        private const string REMOVE_MESSAGE_SCRIPT = "@RemoveMessage";
        private const string START_PROCESS_SCRIPT = "@StartProcess";

        private bool _isDisposed;
        private readonly IDirectory _directory;
        private readonly FlowerClientSettings _settings;
        private readonly CallbackQueue _callbackQueue;
        private readonly ServiceHost _callbackHost;

        public IDirectory Directory
        {
            get { return _directory; }
        }

        #region Constructor

        public FlowerClient(IDirectory directory)
            : this(directory, new FlowerClientSettings())
        {
        }

        /// <summary>
        /// Constructs the client and attaches the handler to 
        /// the assembly resolve event of the current application domain.
        /// </summary>
        /// <param name="directory">The directory to use.</param>
        /// <param name="settings">The client's settings.</param>
        public FlowerClient(IDirectory directory, FlowerClientSettings settings)
        {
            if (directory == null) throw new ArgumentNullException("directory");

            _directory = directory;
            _settings = settings;
            _threadStopEvents = new ManualResetEventSlim[1];

            StartAppContextsManagers();

            _appContextExpirationManagerTimer = new Timer
            (
                state => 
                    _appContextExpirationManagerCommands.Add
                    (
                        new AppContextExpirationManagerCommand 
                        {
                            Action = AppContextExpirationManagerAction.Refresh
                        }
                    ), 
                null, 
                settings.ApplicationContextExpirationCheckInterval,
                settings.ApplicationContextExpirationCheckInterval
            );

            if (settings.EnableCallbacks)
            {
                _callbackQueue = new CallbackQueue();
                _callbackHost = new ServiceHost(_callbackQueue);
                _callbackHost.Open();
            }
            else
            {
                _callbackQueue = null;
                _callbackHost = null;
            }

            AppDomain.CurrentDomain.AssemblyResolve += CurrentDomainOnAssemblyResolve;
        }

        #endregion

        #region Assembly resolving

        private Assembly CurrentDomainOnAssemblyResolve(object sender, ResolveEventArgs args)
        {
            try
            {
                var asmName = new AssemblyName(args.Name);
                if (asmName.Name.EndsWith(".resources"))
                {
                    return null;
                }

                Log.Trace(() => string.Format("Resolving assembly {0}.", args.Name));

                foreach (Assembly asm in AppDomain.CurrentDomain.GetAssemblies())
                {
                    if (asm.GetName().Name == asmName.Name)
                    {
                        Log.Trace(() => string.Format("Assembly {0} found for the name {1}.", asm.FullName, args.Name));
                        return asm;
                    }
                }

                Log.Trace(() => string.Format("Loading assembly {0} from the directory.", asmName.Name));

                string assemblyPath = DirectoryUtil.ResolveAssemblyName(args.Name);

                var scriptResult = CreateScript()
                    .SetText("@GetResolveLinks")
                    .AddParameter("path", assemblyPath)
                    .Execute();

                var assembly = scriptResult.Get<Services.Data.Assembly>("result");

                return assembly != null ? Assembly.Load(assembly.Image) : null;
            }
            catch(Exception ex)
            {
                Log.Fatal(string.Format("Failed to resolve assembly {0}.", args.Name), ex);
                throw;
            }
        }

        #endregion

        #region Service containers management

        private readonly ReaderWriterLockSlim _globalAppContextLock = new ReaderWriterLockSlim();
        private IApplicationContext _globalAppContext;
        private IList<DirectoryEntry> _globalServiceContainers = new List<DirectoryEntry>();
        private readonly ConcurrentDictionary<string, AppContextContainer> _appContexts =
            new ConcurrentDictionary<string, AppContextContainer>();

        private struct AppContextContainer
        {
            public string Id;
            public IApplicationContext AppContext;
            public Exception Exception;
            public DateTime LastUpdatedOn;
        }

        private enum AppContextExpirationManagerAction
        {
            Register,
            Refresh,
            Clear,
            Stop
        }

        private struct AppContextExpirationManagerCommand
        {
            public AppContextExpirationManagerAction Action;
            public string ServiceContainerPath;
            public AppContextContainer AppContext;
        }

        private IActor<LoadAppContextMessage> _appContextsLoader;
        private readonly BlockingCollection<AppContextExpirationManagerCommand> _appContextExpirationManagerCommands =
            new BlockingCollection<AppContextExpirationManagerCommand>();
        private readonly Timer _appContextExpirationManagerTimer;
        private readonly ManualResetEventSlim[] _threadStopEvents;

        #region Start/Stop

        private void StartAppContextsManagers()
        {
            _threadStopEvents[0] = new ManualResetEventSlim(false);
            ThreadPool.QueueUserWorkItem(AppContextsExpirationManager, 0);

            _appContextsLoader = new MultiActor<AppContextsLoader, LoadAppContextMessage>
            (
                _settings.ApplicationContextLoadersCount,
                i => new Actor<AppContextsLoader, LoadAppContextMessage>(new AppContextsLoader(this))
            );
        }

        private void StopAppContextsManagers()
        {
            _appContextsLoader.Dispose();

            _appContextExpirationManagerCommands.Add
            (
                new AppContextExpirationManagerCommand
                {
                    Action = AppContextExpirationManagerAction.Stop
                }
            );

            foreach (ManualResetEventSlim ev in _threadStopEvents)
            {
                if (!ev.Wait(TimeSpan.FromSeconds(30)))
                {
                    Log.Error("Failed to stop thread correctly.");
                }
            }

            Log.Trace("All directory client threads are stopped.");
        }

        #endregion

        private static byte[] GetServiceContainerData(DirectoryEntry directoryEntry)
        {
            var serviceContainer = directoryEntry as ServiceContainer;

            if (serviceContainer != null)
            {
                return serviceContainer.Configuration.Data;
            }

            var processor = directoryEntry as Processor;

            if (processor != null)
            {
                if (processor.Configuration == null)
                {
                    throw new InvalidOperationException(string.Format("Processor {0} has no configuration.", processor.Name));
                }

                return processor.Configuration.Data;
            }

            throw new ArgumentException(string.Format("Directory entry type {0} is not supported as a services container.", directoryEntry.GetType().Name));
        }

        private static IApplicationContext ServiceContainersToAppContext(IApplicationContext parentContext, IEnumerable<DirectoryEntry> serviceContainers)
        {
            return new XmlApplicationContext
            (
                new XmlApplicationContextArgs
                {
                    ParentContext = parentContext,
                    ConfigurationResources = 
                        serviceContainers
                            .Select
                            (
                                serviceContainer => 
                                    (IResource)new InputStreamResource
                                    (
                                        new MemoryStream(GetServiceContainerData(serviceContainer)), 
                                        serviceContainer.Name
                                    )
                            )
                            .ToArray()
                }
            );
        }

        #region AppContextsLoader

        private class LoadAppContextMessage : IMessage<AppContextsLoader>
        {
            public string ServiceContainerPath;
            public BlockingCollection<AppContextContainer> AppContext;

            public void Process(AppContextsLoader processor)
            {
                processor.Process(this);
            }
        }

        private class AppContextsLoader
        {
            private readonly FlowerClient _client;

            public AppContextsLoader(FlowerClient client)
            {
                _client = client;
            }

            public void Process(LoadAppContextMessage message)
            {
                try
                {
                    message.AppContext.Add
                    (
                        _client._appContexts.GetOrAdd(message.ServiceContainerPath, LoadAppContext)
                    );
                }
                catch (IndexOutOfRangeException)
                {
                    message.AppContext.Add
                    (
                        new AppContextContainer { AppContext = null }
                    );
                }
                catch (Exception ex)
                {
                    message.AppContext.Add
                    (
                        new AppContextContainer { Exception = ex }
                    );
                }
            }

            private AppContextContainer LoadAppContext(string serviceContainerPath)
            {
                Log.Trace(() => string.Format("Loading service container '{0}' from the directory.", serviceContainerPath));

                _client._globalAppContextLock.EnterUpgradeableReadLock();
                try
                {
                    IList<DirectoryEntry> globalServiceContainers = null;
                    DirectoryEntry serviceContainer = null;

                    bool useGlobalContext;

                    if (serviceContainerPath.StartsWith("/Services"))
                    {
                        useGlobalContext = true;

                        var trnOpt = new TransactionOptions
                        {
                            IsolationLevel = IsolationLevel.ReadCommitted,
                            Timeout = TimeSpan.FromMinutes(1)
                        };

                        using (var trn = new TransactionScope(TransactionScopeOption.Required, trnOpt))
                        {
                            Script.Result requestResult =
                                _client.CreateScript()
                                    .SetText("@GetServiceContainer")
                                    .AddParameter("serviceContainerPath", serviceContainerPath)
                                    .AddParameter("globalServiceContainers", _client._globalServiceContainers)
                                    .Execute();

                            globalServiceContainers =
                                requestResult.Get<IList<DirectoryEntry>>("globalServiceContainers");
                            serviceContainer =
                                requestResult.Get<DirectoryEntry>("serviceContainer");

                            trn.Complete();
                        }
                    }
                    else
                    {
                        useGlobalContext = false;
                        serviceContainer = _client.Directory.Get(serviceContainerPath);
                    }

                    if (globalServiceContainers != null)
                    {
                        Log.Trace("Updating global app. context.");

                        _client._globalAppContextLock.EnterWriteLock();
                        try
                        {
                            _client._globalAppContext =
                                ServiceContainersToAppContext
                                (
                                    null,
                                    globalServiceContainers
                                        .Select(svcnt => (ServiceContainer)svcnt)
                                );

                            foreach (DirectoryEntry globalServiceContainer in globalServiceContainers)
                            {
                                ((ServiceContainer)globalServiceContainer).Configuration = null;
                                globalServiceContainer.UpdatedOn += TimeSpan.FromMilliseconds(1);
                            }

                            _client._globalServiceContainers = globalServiceContainers;

                            _client._appContexts.Clear();
                            _client._appContextExpirationManagerCommands.Add
                            (
                                new AppContextExpirationManagerCommand
                                {
                                    Action = AppContextExpirationManagerAction.Clear
                                }
                            );
                        }
                        finally
                        {
                            _client._globalAppContextLock.ExitWriteLock();
                        }
                    }

                    if (serviceContainer != null)
                    {
                        var container =
                            new AppContextContainer
                            {
                                Id = serviceContainer.Id,
                                LastUpdatedOn = serviceContainer.UpdatedOn,
                                AppContext = ServiceContainersToAppContext
                                (
                                    (useGlobalContext) ? _client._globalAppContext : null,
                                    new [] { serviceContainer }
                                )
                            };

                        _client._appContextExpirationManagerCommands.Add
                        (
                            new AppContextExpirationManagerCommand
                            {
                                Action = AppContextExpirationManagerAction.Register,
                                ServiceContainerPath = serviceContainerPath,
                                AppContext = container
                            }
                        );

                        return container;
                    }
                    else
                    {
                        Log.Trace(() => string.Format("Service container '{0}' has not been found.", serviceContainerPath));
                        throw new IndexOutOfRangeException("Service container has not been found.");
                    }
                }
                finally
                {
                    _client._globalAppContextLock.ExitUpgradeableReadLock();
                }
            }
        }

        #endregion

        #region AppContextsExpirationManager

        private void AppContextsExpirationManager(object arg)
        {
            var waitHandleIndex = (int)arg;
            var watchedServiceContainers = new List<ServiceContainer>();
            var idToPathMap = new Dictionary<string, string>();
            var lastRefreshedOn = DateTime.MinValue;

            while (true)
            {
                var command = _appContextExpirationManagerCommands.Take();

                try
                {
                    switch (command.Action)
                    {
                        case AppContextExpirationManagerAction.Register:
                            Log.Trace(() => string.Format("Registering service container '{0}' in the expiration check list.", command.ServiceContainerPath));

                            watchedServiceContainers.Add
                            (
                                new ServiceContainer
                                {
                                    Id = command.AppContext.Id,
                                    UpdatedOn = command.AppContext.LastUpdatedOn
                                }
                            );
                            idToPathMap[command.AppContext.Id] = command.ServiceContainerPath;
                            break;
                        case AppContextExpirationManagerAction.Refresh:
                            if (DateTime.Now - lastRefreshedOn >= _settings.ApplicationContextExpirationCheckInterval)
                            {
                                Log.Trace("Checking service containers for expiration.");

                                var requestResult =
                                    CreateScript()
                                        .SetText("@GetExpiredServiceContainersIds")
                                        .AddParameter("serviceContainers", watchedServiceContainers.Union(_globalServiceContainers))
                                        .Execute();

                                var expiredServiceContainersIds =
                                    requestResult.Get<IList<string>>("expiredServiceContainersIds");

                                if (expiredServiceContainersIds != null)
                                {
                                    if (_globalServiceContainers.Any(gsc => expiredServiceContainersIds.Contains(gsc.Id)))
                                    {
                                        Log.Trace("One of the global service containers expired. Removing all service containers from the expiration check list.");

                                        _appContexts.Clear();
                                        watchedServiceContainers.Clear();
                                        idToPathMap.Clear();
                                    }
                                    else
                                    {
                                        foreach (string id in expiredServiceContainersIds)
                                        {
                                            string path;

                                            if (idToPathMap.TryGetValue(id, out path))
                                            {
                                                Log.Trace(() => string.Format("Service container '{0}' has expired.", path));

                                                AppContextContainer container;
                                                if (_appContexts.TryRemove(path, out container))
                                                {
                                                    watchedServiceContainers.RemoveAll(svcnt => svcnt.Id == id);
                                                    idToPathMap.Remove(id);
                                                }
                                            }
                                        }
                                    }
                                }

                                lastRefreshedOn = DateTime.Now;
                            }
                            break;
                        case AppContextExpirationManagerAction.Clear:
                            Log.Trace("Removing all service containers from the expiration check list.");

                            watchedServiceContainers.Clear();
                            idToPathMap.Clear();
                            break;
                        case AppContextExpirationManagerAction.Stop:
                            Log.Trace("App. context expiration manager is stopped.");
                            _threadStopEvents[waitHandleIndex].Set();
                            return;
                        default:
                            throw new ArgumentOutOfRangeException(String.Format("Unknown action {0}.", command.Action));
                    }
                }
                catch (Exception ex)
                {
                    Log.Error("Error ocured in app. context expiration manager.", ex);
                }
            }
        }

        #endregion

        public TSvc GetService<TSvc>(string servicePath)
        {
            if (string.IsNullOrEmpty(servicePath))
            {
                throw new ArgumentException("Service path has not been provided.", "servicePath");
            }

            if (!servicePath.StartsWith("/"))
            {
                servicePath = "/Services/" + servicePath;
            }

            if (servicePath.StartsWith("/Services/Flower/", StringComparison.Ordinal))
            {
                throw new InvalidOperationException("Service containers in '/Services/Flower' are not accessible directly.");
            }

            string serviceContainerPath, serviceId;

            DirectoryUtil.CutPathTailSegment(servicePath, out serviceContainerPath, out serviceId);

            if (String.IsNullOrEmpty(serviceContainerPath))
            {
                throw new ArgumentException("Service path doesn't contain a service container path.", "servicePath");
            }

            if (String.IsNullOrEmpty(serviceId))
            {
                throw new ArgumentException("Service path doesn't contain a service code within the container.", "servicePath");
            }

            AppContextContainer container;
            if (_appContexts.TryGetValue(serviceContainerPath, out container))
            {
                return GetServiceInstance<TSvc>(container.AppContext, serviceContainerPath, serviceId);
            }

            var appContext = new BlockingCollection<AppContextContainer>();
            _appContextsLoader.Send
            (
                new LoadAppContextMessage
                {
                    ServiceContainerPath = serviceContainerPath,
                    AppContext = appContext
                }
            );

            container = appContext.Take();

            if (container.Exception != null)
            {
                throw new ServicesProviderException(container.Exception);
            }

            if (container.AppContext == null)
            {
                throw new ServicesProviderException(string.Format("Service container {0} doesn't exist.", serviceContainerPath));
            }

            return GetServiceInstance<TSvc>(container.AppContext, serviceContainerPath, serviceId);
        }

        private TSvc GetServiceInstance<TSvc>(IApplicationContext appContext, string serviceContainerPath, string serviceId)
        {
            object svc;
            try
            {
                svc = appContext.GetObject(serviceId);

                if (svc is TSvc)
                {
                    if (typeof(TSvc).IsInterface && appContext.IsSingleton(serviceId))
                    {
                        string servicePath = string.Concat(serviceContainerPath, "/", serviceId);
                        Log.Trace(() => string.Format("Service {0} is wrapped into an autorefreshing proxy.", servicePath));

                        return
                            (TSvc)new ProxyFactory
                            (
                                typeof(TSvc), 
                                new AutorefreshingProxyInterceptor
                                (
                                    this,
                                    servicePath, 
                                    _settings.ApplicationContextExpirationCheckInterval
                                )
                            )
                            .GetProxy();
                    }
                    else
                    {
                        return (TSvc)svc;
                    }
                }
                else
                {
                    throw new ServicesProviderException
                    (
                        string.Format
                        (
                            "Service {0}/{1} has type {2} which is not convertible to type {3}.",
                            serviceContainerPath,
                            serviceId,
                            svc.GetType().AssemblyQualifiedName,
                            typeof(TSvc).AssemblyQualifiedName
                        )
                    );
                }
            }
            catch (NoSuchObjectDefinitionException)
            {
                throw new ServicesProviderException(string.Format("Service {0} is not defined in the container {1}.", serviceId, serviceContainerPath));
            }
            catch (Exception ex)
            {
                throw new ServicesProviderException(ex);
            }
        }

        public IQueueService GetProcessQueue(string pid)
        {
            string processPath;

            using (var txn = new TransactionScope(TransactionScopeOption.Suppress))
            {
                processPath = _directory.GetPathById(pid);
            }

            string[] pathSegments = DirectoryUtil.SplitPath(processPath);

            return GetService<IQueueService>(string.Concat("/Processors/", pathSegments[1], "/queue"));
        }

        #endregion

        #region Processes and workflows

        private static void StartProcess
        (
            IFlowerClient client, 
            string workflow, 
            DirectoryEntry newProcess, 
            DirectoryEntry initialState, 
            IEnumerable<DirectoryEntry> localSets, 
            out string pid, 
            out string processorName
        )
        {
            var result =
                client
                    .CreateScript()
                    .SetText(START_PROCESS_SCRIPT)
                    .AddParameter("workflow", workflow)
                    .AddParameter("newProcess", newProcess)
                    .AddParameter("initialState", initialState)
                    .AddParameter("localSets", localSets)
                    .Execute();

            pid = result.Get<string>("pid");
            processorName = result.Get<string>("processorName");
        }

        private class SetsBuilder : ISetsBuilder
        {
            public readonly List<Set> Sets = new List<Set>(); 

            public ISetsBuilder Set
            (
                string name, 
                Type messageType,
                int capacity = 0,
                Func<ISetPermissionBuilder, ISetPermissionBuilder> acl = null
            )
            {
                if (name == "Log")
                {
                    return this;
                }

                if (Sets.Any(s => s.Name == name))
                {
                    throw new InvalidOperationException(string.Format("Local set {0} is already defined.", name));
                }
                
                Sets.Add(new Set
                {
                    Name = name,
                    MessageType = messageType.AssemblyQualifiedName,
                    Capacity = capacity,
                    AccessControlList = (acl != null) ? ((SetPermissionBuilder)acl(new SetPermissionBuilder())).Acl : new List<AccessControlEntry>()
                });

                return this;
            }
        }

        private static IEnumerable<Set> GetLocalSets<TIn, TOut>(IWorkflow<TIn, TOut> wfDefinition)
        {
            var bld = new SetsBuilder();
            wfDefinition.DefineLocalSets(bld);

            foreach (Set set in bld.Sets)
            {
                yield return set;
            }

            yield return new Set
            {
                Name = "Log",
                MessageType = typeof(string).AssemblyQualifiedName
            };
        }

        public void InjectServices(object target)
        {
            Type wfType = target.GetType();

            foreach (PropertyInfo propertyInfo in wfType.GetProperties(BindingFlags.Public | BindingFlags.NonPublic | BindingFlags.SetProperty | BindingFlags.Instance))
            {
                if (propertyInfo.GetIndexParameters().Length == 0)
                {
                    if (propertyInfo.PropertyType == typeof(IDirectory))
                    {
                        propertyInfo.SetValue(target, _directory);
                        continue;
                    }

                    TryInjectService(target, propertyInfo);
                }
            }

            foreach (FieldInfo fieldInfo in wfType.GetFields(BindingFlags.Public | BindingFlags.NonPublic | BindingFlags.SetField | BindingFlags.Instance))
            {
                if (fieldInfo.FieldType == typeof(IDirectory))
                {
                    fieldInfo.SetValue(target, _directory);
                    continue;
                }

                TryInjectService(target, fieldInfo);
            }
        }

        private void TryInjectService(object target, MemberInfo targetMember)
        {
            object[] serviceAttrs = targetMember.GetCustomAttributes(typeof(ServiceAttribute), false);

            if (serviceAttrs.Length > 0)
            {
                var serviceAttr = serviceAttrs[0] as ServiceAttribute;

                if (serviceAttr != null)
                {
                    string fullServicePath =
                        (serviceAttr.ServicePath.StartsWith("/")) ?
                            serviceAttr.ServicePath :
                            "/Services/" + serviceAttr.ServicePath;

                    var svc = GetService<object>(fullServicePath);

                    if (svc != null)
                    {
                        try
                        {
                            if (targetMember is FieldInfo)
                            {
                                ((FieldInfo)targetMember)
                                    .SetValue(target, svc);
                            }
                            else if (targetMember is PropertyInfo)
                            {
                                ((PropertyInfo)targetMember)
                                    .SetValue(target, svc);
                            }
                        }
                        catch (Exception ex)
                        {
                            Log.Warning(string.Format("Failed to inject the service '{0}': {1}.", fullServicePath, ex.Message));
                        }
                    }
                    else
                    {
                        Log.Warning(string.Format("Failed to inject the service '{0}': service not found.", fullServicePath));
                    }
                }
            }
        }

        public ProcessStartResult<TOut> StartProcess<TIn, TOut>(string workflow, TIn args, Log log)
        {
            if (workflow == null) throw new ArgumentNullException("workflow");
            if (log == null) throw new ArgumentNullException("log");

            if (!workflow.StartsWith("#") && !workflow.StartsWith("/"))
            {
                workflow = "/Workflows/" + workflow;
            }

            var wf = _directory.Get(workflow) as Services.Data.Workflow;

            if (wf == null)
            {
                throw new ArgumentException(string.Format("The workflow '{0}' doesn't exist.", workflow), "workflow");
            }

            var wfDefinitionType = Type.GetType(wf.DefinitionType);
            if (wfDefinitionType == null)
            {
                throw new TypeLoadException(string.Format("Cannot load type {0}.", wf.DefinitionType));
            }

            var wfDefinitionTypeConstructor = wfDefinitionType.GetConstructor(Type.EmptyTypes);
            if (wfDefinitionTypeConstructor == null)
            {
                throw new InvalidOperationException(string.Format("The type {0} has no parameterless constructor.", wf.DefinitionType));
            }

            var wfDefinition = (IWorkflow<TIn, TOut>)wfDefinitionTypeConstructor.Invoke(EmptyObjectArray);

            InjectServices(wfDefinition);

            var transOpts = 
                new TransactionOptions 
                {
                    IsolationLevel = IsolationLevel.ReadCommitted, 
                    Timeout = _settings.TransactionTimeout
                };
            using (var trn = new TransactionScope(TransactionScopeOption.Required, transOpts)) 
            {
                var result = 
                    new ProcessStartResult<TOut> 
                    {
                        Result = wfDefinition.Initialize(new InitializationContext(this, log), args)
                    };

                var process = 
                    new Process
                    {
                        Workflow = wf.DefinitionType
                    };

                var stateEntry = new State
                {
                    Name = Guid.NewGuid().ToString(),
                    Data = wfDefinition.ToBlob(BlobFormat.GZipXml, true)
                };

                string pid;
                string processorName;
                StartProcess
                (
                    this, 
                    workflow, 
                    process, 
                    stateEntry,
                    GetLocalSets(wfDefinition), 
                    out pid, 
                    out processorName
                );

                if (pid == null)
                {
                    throw new InvalidOperationException("Failed to schedule a process. No processors available.");
                }

                result.Pid = pid;
                result.ProcessorName = processorName;

                trn.Complete();

                return result;
            }
        }

        #endregion

        #region Messaging

        public IList<Message<TMsg>> GetMessages<TMsg>(string subset, SortProperty property, SortOrder order, long offset, long limit)
        {
            IList<DirectoryEntry> messages =
                _directory.GetChildren(subset, DirectoryEntryTypes.Message, property, order, offset, limit);

            return
                messages
                    .Select(msg => MessageEntryToMessage<TMsg>((Message)msg))
                    .ToList();
        }

        public Message<TMsg> GetMessage<TMsg>(string message)
        {
            var msg = _directory.Get(message) as Message;

            return msg != null ? MessageEntryToMessage<TMsg>(msg) : null;
        }

        private static Message<TMsg> MessageEntryToMessage<TMsg>(Message message)
        {
            return new Message<TMsg>
            {
                Id = message.Id,
                Name = message.Name,
                Data = message.Data.To<TMsg>()
            };            
        }

        public bool PutMessage<TMsg>(string subset, string name, TMsg message, BlobFormat format)
        {
            var transOpts =
                new TransactionOptions
                {
                    IsolationLevel = IsolationLevel.ReadCommitted,
                    Timeout = _settings.TransactionTimeout
                };

            using (var trn = new TransactionScope(TransactionScopeOption.Required, transOpts))
            {
                bool result =
                    ScriptHelper.Enqueue
                    (
                        this,
                        subset,
                        new Message
                        {
                            Name = name,
                            Data = message.ToBlob(format)
                        },
                        null
                    );
                trn.Complete();
                return result;
            }
        }

        public int PutMessages<TMsg>(string subset, IList<string> names, IList<TMsg> messages, BlobFormat format)
        {
            var transOpts =
                new TransactionOptions
                {
                    IsolationLevel = IsolationLevel.ReadCommitted,
                    Timeout = _settings.TransactionTimeout
                };

            using (var trn = new TransactionScope(TransactionScopeOption.Required, transOpts))
            {
                int result =
                    ScriptHelper.EnqueueMultiple
                    (
                        this,
                        subset,
                        messages.Select
                        (
                            (m, i) => 
                                (DirectoryEntry)new Message 
                                {
                                    Name = (names != null) ? names[i] : null,
                                    Data = m.ToBlob(format)
                                }
                        ).ToList()
                    );
                trn.Complete();
                return result;
            }
        }

        public void RemoveMessage(string message)
        {
            CreateScript()
                .SetText(REMOVE_MESSAGE_SCRIPT)
                .AddParameter("message", message)
                .Execute();
        }

        #endregion

        #region Locking

        public LockFailure[] Lock(IEnumerable<string> resources, string owner)
        {
            Transaction currentTransaction = Transaction.Current;
            var transOpts =
                new TransactionOptions
                {
                    IsolationLevel =
                        (currentTransaction != null) ?
                        currentTransaction.IsolationLevel :
                        IsolationLevel.ReadCommitted,
                    Timeout = _settings.TransactionTimeout
                };

            using (var trn = new TransactionScope(TransactionScopeOption.Required, transOpts))
            {
                LockFailure[] result = ScriptHelper.Lock(this, resources, owner, null);
                trn.Complete();
                return result;
            }
        }

        public void Unlock(IEnumerable<string> resources, string owner)
        {
            ScriptHelper.Unlock(this, resources, owner);
        }

        #endregion

        #region Scripting

        public Script CreateScript()
        {
            return new Script(_directory);
        }

        #endregion

        #region Callbacks

        public void AddRecipient<TMsg>(string recipient, Action<TMsg> handler)
        {
            if (_callbackQueue == null) throw new InvalidOperationException("Callbacks are disabled.");
            if (recipient == null) throw new ArgumentNullException("recipient");
            if (handler == null) throw new ArgumentNullException("handler");

            _callbackQueue
                .AddRecipient
                (
                    recipient,
                    blob =>
                    {
                        Log.Trace(() => string.Format("Received message {0} at {1}.", typeof(TMsg).FullName, recipient));

                        try
                        {
                            handler(blob.To<TMsg>());
                        }
                        catch(Exception ex)
                        {
                            Log.Error(string.Format("Failed to handle message {0} at {1}.", typeof(TMsg).FullName, recipient), ex);
                        }
                    }
                );
        }

        public void RemoveRecipient(string recipient)
        {
            if (_callbackQueue == null) throw new InvalidOperationException("Callbacks are disabled.");
            if (recipient == null) throw new ArgumentNullException("recipient");

            _callbackQueue.RemoveRecipient(recipient);
        }

        #endregion

        #region Disposing

        /// <summary>
        /// Disposes the client.
        /// </summary>
        public void Dispose()
        {
            if (!_isDisposed)
            {
                if (_callbackHost != null)
                {
                    _callbackHost.Close();
                }

                _appContextExpirationManagerTimer.Dispose();

                StopAppContextsManagers();

                AppDomain.CurrentDomain.AssemblyResolve -= CurrentDomainOnAssemblyResolve;

                _isDisposed = true;
            }

            GC.SuppressFinalize(this);
        }

        #endregion
    }
}
